package iterator;

import heap.*;
import global.*;
import btree.Interface3;
import bufmgr.*;
import diskmgr.*;
import index.*;
import java.lang.*;
import java.io.*;

/**
 * 
 * This file contains an implementation of the nested loops join algorithm as
 * described in the Shapiro paper. The algorithm is extremely simple:
 * 
 * foreach tuple r in R do foreach tuple s in S do if (ri == sj) then add (r, s)
 * to the result.
 */

public class NestedLoopsJoins extends Iterator {
	private AttrType _in1[], _in2[];
	private int in1_len, in2_len;
	private Iterator outer; // this has to be Iterator
	private short t2_str_sizescopy[];
	private CondExpr OutputFilter[];
	private CondExpr RightFilter[];
	private int n_buf_pgs; // # of buffer pages available.
	private boolean done, // Is the join complete
			get_from_outer; // if TRUE, a tuple is got from outer
	private Tuple outer_tuple, inner_tuple;
	private Tuple Jtuple; // Joined tuple
	private FldSpec perm_mat[];
	private int nOutFlds;
	private Heapfile hf;
	private Scan inner;
	private RID _rid = new RID();
	private Interface3 i = null;
	public int myctr = 0;
	public int BPjoinNodePosition;
	public int JoinOnSubjectorObject;
	public Tuple tr_out ;
	public Interface3 i3;
	public AttrType [] JTupleAttr;
	public Float confFilter;
	
	/**
	 * constructor Initialize the two relations which are joined, including
	 * relation type,
	 * 
	 * @param in1
	 *            Array containing field types of R.
	 * @param len_in1
	 *            # of columns in R.
	 * @param t1_str_sizes
	 *            shows the length of the string fields.
	 * @param in2
	 *            Array containing field types of S
	 * @param len_in2
	 *            # of columns in S
	 * @param t2_str_sizes
	 *            shows the length of the string fields.
	 * @param amt_of_mem
	 *            IN PAGES
	 * @param am1
	 *            access method for left i/p to join
	 * @param relationName
	 *            access hfapfile for right i/p to join
	 * @param outFilter
	 *            select expressions
	 * @param rightFilter
	 *            reference to filter applied on right i/p
	 * @param proj_list
	 *            shows what input fields go where in the output tuple
	 * @param n_out_flds
	 *            number of outer relation fileds
	 * @exception IOException
	 *                some I/O fault
	 * @exception NestedLoopException
	 *                exception from this class
	 */
	public NestedLoopsJoins(AttrType in1[], int len_in1, short t1_str_sizes[],
			AttrType in2[], int len_in2, short t2_str_sizes[],
			int amt_of_mem,
			Iterator am1, // this is supposed to be Iterator object.
			String relationName, CondExpr outFilter[], CondExpr rightFilter[],
			FldSpec proj_list[], int n_out_flds, Interface3 i,int BPjoinNodePosition, int JoinOnSubjectorObject,
			Interface3 i3,Float confidenceFilter)
			throws IOException, NestedLoopException {
		this.i = i;
		_in1 = new AttrType[in1.length];
		_in2 = new AttrType[in2.length];
		System.arraycopy(in1, 0, _in1, 0, in1.length);
		System.arraycopy(in2, 0, _in2, 0, in2.length);
		in1_len = len_in1;
		in2_len = len_in2;
		this.BPjoinNodePosition=BPjoinNodePosition;
		this.JoinOnSubjectorObject=JoinOnSubjectorObject;
		myctr = 0;
		confFilter = confidenceFilter;
		outer = am1;
		t2_str_sizescopy = t2_str_sizes;
		inner_tuple = new Tuple();
		Jtuple = new Tuple();
		OutputFilter = outFilter;
		RightFilter = rightFilter;
		this.i3 = i3;
		n_buf_pgs = amt_of_mem;
		inner = null;
		done = false;
		get_from_outer = true;

		AttrType[] Jtypes = new AttrType[n_out_flds];
		short[] t_size;
		JTupleAttr = Jtypes;
		perm_mat = proj_list;
		nOutFlds = n_out_flds;
		try {
			t_size = TupleUtils.setup_op_tuple(Jtuple, Jtypes, in1, len_in1,
					in2, len_in2, t1_str_sizes, t2_str_sizes, proj_list,
					nOutFlds);
		} catch (TupleUtilsException e) {
			throw new NestedLoopException(e,
					"TupleUtilsException is caught by NestedLoopsJoins.java");
		}

		try {
			hf = new Heapfile(relationName);

		} catch (Exception e) {
			throw new NestedLoopException(e, "Create new heapfile failed.");
		}
	}

	/**
	 * @return The joined tuple is returned
	 * @exception IOException
	 *                I/O errors
	 * @exception JoinsException
	 *                some join exception
	 * @exception IndexException
	 *                exception from super class
	 * @exception InvalidTupleSizeException
	 *                invalid tuple size
	 * @exception InvalidTypeException
	 *                tuple type not valid
	 * @exception PageNotReadException
	 *                exception from lower layer
	 * @exception TupleUtilsException
	 *                exception from using tuple utilities
	 * @exception PredEvalException
	 *                exception from PredEval class
	 * @exception SortException
	 *                sort exception
	 * @exception LowMemException
	 *                memory error
	 * @exception UnknowAttrType
	 *                attribute type unknown
	 * @exception UnknownKeyTypeException
	 *                key type unknown
	 * @exception Exception
	 *                other exceptions
	 */
	public Tuple get_next() throws IOException, JoinsException, IndexException,
			InvalidTupleSizeException, InvalidTypeException,
			PageNotReadException, TupleUtilsException, PredEvalException,
			SortException, LowMemException, UnknowAttrType,
			UnknownKeyTypeException, Exception {
		// This is a DUMBEST form of a join, not making use of any key
		// information...
		
		if (done)
			return null;

		do {
			
			// If get_from_outer is true, Get a tuple from the outer, delete
			// an existing scan on the file, and reopen a new scan on the file.
			// If a get_next on the outer returns DONE?, then the nested loops
			// join is done too.

			if (get_from_outer == true) {
				tr_out = new Tuple();
				tr_out.setHdr((short) _in1.length, _in1, null);

				get_from_outer = false;
				if (inner != null) // If this not the first time,
				{
					// close scan
					inner = null;
				}

				try {
					inner = hf.openScan();

				} catch (Exception e) {
					throw new NestedLoopException(e, "openScan failed");
				}

				// if(outer.get_next()!=null){
				// System.out.println("iterator has records");
				// }
				boolean isTrue = false;
				boolean flag = true;
				if(outer instanceof IndexScan){
					IndexScan iscan = (IndexScan) outer;
					while (flag) {
						outer_tuple = iscan.get_next(_rid, true);
						if(Interface3.isRIDModified(_rid)) {
							isTrue = true;
							Tuple temp = i.tripleHeap.getRecord(_rid);
							tr_out.tupleCopy(temp);
							float conf = tr_out.getFloFld(_in1.length);
							//System.out.println("Outer Table");
							//tr_out.print(_in1);
							//i3.lifeLine(tr_out, _in1);
							//System.out.println("Checking " + t.getConfidence());
							//System.out.println("Filter cond:" + i3.bp_conf_filter);
							if (conf > confFilter) {
								flag = false;
								//System.out.println("You are good. Go ahead");								
							} else {
								//System.out.println("You didn't pass! sorry bro");
							}
							
						}else {
							//Loop is exhausted.
							flag = false;
						}
					}
				}else if(outer instanceof NestedLoopsJoins){
					NestedLoopsJoins nscan = (NestedLoopsJoins) outer;
					tr_out = nscan.get_next();
					
					
					if(tr_out != null){
						//System.out.println("Outer Table");
						//tr_out.print(_in1);
						//i3.lifeLine(tr_out, _in1);
						isTrue = true;
					}
				}
				if (!isTrue) {

					done = true;
					if (inner != null) {

						inner = null;
					}

					return null;
				}
				
				//System.out.println("OUTER DATA");
				//tr_out.printTuple();
				// System.out.println("RID:" + _rid);
				// System.out.println("OUTER PAGE ID: "+temp.getFldCnt());//
				// .getIntFld(1)+" , OUTER SLOT NO: "+outer_tuple.getIntFld(2));
			} // ENDS: if (get_from_outer == TRUE)

			// The next step is to get a tuple from the inner,
			// while the inner is not completely scanned && there
			// is no match (with pred),get a tuple from the inner.

			RID rid = new RID();
			while ((inner_tuple = inner.getNext(rid)) != null) {
				// System.out.println("inner file reading going on.");
				// Tuple temp = i.tripleHeap.getRecord(_rid);
				Triple tr_in = new Triple();
				tr_in.tupleCopy(inner_tuple);
				
                //System.out.println("inner tuple"+i3.cheatingCounter);
				//tr_in.printTuple();
				try {
                //i3.lifeLine(inner_tuple, _in2);
				} catch (Exception e) {
					int a = 1;
				}
				if (PredEval.Eval(RightFilter, tr_in, null, _in2, null) == true) {
					if (PredEval.Eval(OutputFilter, tr_out, tr_in, _in1, _in2) == true) {
						// Apply a projection on the outer and inner tuples.
						Projection.Join(tr_out, _in1, tr_in, _in2, Jtuple,
								perm_mat, nOutFlds);
//						System.out.println("no_out_flds="+nOutFlds);
//						BasicPattern bp = new BasicPattern((nOutFlds/2);
//						bp.tupleCopy(Jtuple);
//						i.BPtypes = bp.types;
//						System.out.println("type length is = "+bp.types.length + "no_out_flds="+nOutFlds);
					//	System.out.println("Matched data");
						//Jtuple.print(JTupleAttr);
						//i3.lifeLine(Jtuple, JTupleAttr);
						return Jtuple;						
					}
				}				
			}
			inner.closescan();

			// There has been no match. (otherwise, we would have
			// returned from t//he while loop. Hence, inner is
			// exhausted, => set get_from_outer = TRUE, go to top of loop

			get_from_outer = true; // Loop back to top and get next outer tuple.
		} while (true);
	}

	/**
	 * implement the abstract method close() from super class Iterator to finish
	 * cleaning up
	 * 
	 * @exception IOException
	 *                I/O error from lower layers
	 * @exception JoinsException
	 *                join error from lower layers
	 * @exception IndexException
	 *                index access error
	 */
	public void close() throws JoinsException, IOException, IndexException {
		if (!closeFlag) {

			try {
				outer.close();
			} catch (Exception e) {
				throw new JoinsException(e,
						"NestedLoopsJoin.java: error in closing iterator.");
			}
			closeFlag = true;
		}
	}
}


//if(BPjoinNodePosition==1 && JoinOnSubjectorObject==0){
//if (tr_in.getSubjectID().equals(tr_out.getSubjectID())) {
////	System.out.println("MATCH MATCH" + ++this.myctr);
//	//tr_in.printTuple();
//	//tr_out.printTuple();
//	return (new BasicPattern(new NodeID(new PageId(tr_out.getIntFld(1)), tr_out.getIntFld(2)), new NodeID(new PageId(tr_out.getIntFld(5)), tr_out.getIntFld(6)),
//			new NodeID(new PageId(tr_in.getIntFld(1)), tr_in.getIntFld(2)), new NodeID(new PageId(tr_in.getIntFld(5)), tr_in.getIntFld(6)), tr_in.getFloFld(7)));
//}
//
//}
//else if(BPjoinNodePosition==2 && JoinOnSubjectorObject==0){
//	if (tr_in.getObjectID().equals(tr_out.getSubjectID())) {
////		System.out.println("MATCH MATCH" + ++this.myctr);
//		//tr_in.printTuple();
//		//tr_out.printTuple();
//		//float min_conf=min();
//		 return (new BasicPattern  (new NodeID(new PageId(tr_out.getIntFld(1)), tr_out.getIntFld(2)), new NodeID(new PageId(tr_out.getIntFld(5)), tr_out.getIntFld(6)),
//				new NodeID(new PageId(tr_in.getIntFld(1)), tr_in.getIntFld(2)), new NodeID(new PageId(tr_in.getIntFld(5)), tr_in.getIntFld(6)), tr_in.getFloFld(7)));
//	}
//}
//	else if(BPjoinNodePosition==1 && JoinOnSubjectorObject==1){
//		if (tr_in.getSubjectID().equals(tr_out.getObjectID())) {
////			System.out.println("MATCH MATCH" + ++this.myctr);
//			//tr_in.printTuple();
//			//tr_out.printTuple();
//			return (new BasicPattern(new NodeID(new PageId(tr_out.getIntFld(1)), tr_out.getIntFld(2)), new NodeID(new PageId(tr_out.getIntFld(5)), tr_out.getIntFld(6)),
//					new NodeID(new PageId(tr_in.getIntFld(1)), tr_in.getIntFld(2)), new NodeID(new PageId(tr_in.getIntFld(5)), tr_in.getIntFld(6)), tr_in.getFloFld(7)));
//		}
//	}
//	else if(BPjoinNodePosition==2 && JoinOnSubjectorObject==1){
//		if (tr_in.getObjectID().equals(tr_out.getObjectID())) {
////			System.out.println("MATCH MATCH" + ++this.myctr);
//			//tr_in.printTuple();
//			//tr_out.printTuple();
//			return (new BasicPattern(new NodeID(new PageId(tr_out.getIntFld(1)), tr_out.getIntFld(2)), new NodeID(new PageId(tr_out.getIntFld(5)), tr_out.getIntFld(6)),
//					new NodeID(new PageId(tr_in.getIntFld(1)), tr_in.getIntFld(2)), new NodeID(new PageId(tr_in.getIntFld(5)), tr_in.getIntFld(6)), tr_in.getFloFld(7)));
//		}
//	}

// tr_in.printTuple();
//tr_in.setHdr((short) in2_len, _in2, t2_str_sizescopy);
// System.out.println("INNER PAGE ID: "+inner_tuple.getIntFld(1)+" , INNER SLOT ID: "+inner_tuple.getIntFld(2));